package com.xuelang.pipeline.worker.app.log;

import com.xuelang.pipeline.worker.app.pod.PodService;
import com.xuelang.pipeline.worker.client.pod.dto.PodQueryDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import javax.annotation.Resource;
import java.io.IOException;
import java.net.URI;
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * <p>日志提供方 WebSocket 处理器</p>
 *
 * @author : yixin.lys
 * @date : 2024-07-01 10:44
 **/
@Slf4j
public class LogProviderWebSocketHandler extends TextWebSocketHandler {

    private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(10);
    private ScheduledExecutorService heartbeatExecutorService;

    @Resource
    private PodService podService;

    /**
     * 错过心跳次数
     */
    private int missedHeartbeats = 0;
    /**
     * 允许的最大错过次数
     */
    private static final int MAX_MISSED_HEARTBEATS = 3;

    private static final String KUBERNET_CONFIG = "apiVersion: v1\nclusters:\n- cluster:\n    certificate-authority-data: LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSUM1ekNDQWMrZ0F3SUJBZ0lCQURBTkJna3Foa2lHOXcwQkFRc0ZBREFWTVJNd0VRWURWUVFERXdwcmRXSmwKY201bGRHVnpNQjRYRFRJME1EVXlOekF4TkRZek5Wb1hEVE0wTURVeU5UQXhORFl6TlZvd0ZURVRNQkVHQTFVRQpBeE1LYTNWaVpYSnVaWFJsY3pDQ0FTSXdEUVlKS29aSWh2Y05BUUVCQlFBRGdnRVBBRENDQVFvQ2dnRUJBS01iClRoS1Q4R0xCOFFCbzV6eDJQZURDamlWeFVLNk8vVXRYYWhhMzdsbFJmWUtFK1Mva0ZYejhOdEVPbElkcTRVb1UKQktmamk3LzhvZ0kxVEE4eFdMMENZQkt3Q0hxYmtaWUY5eUZVenExNVliOW1MM1pPUVQyQmRjdjhwOGE3Nkd3SwpEU0N6d3ZJT1FIbE81YzJBWlZMWG5nMUlVYlNEcmczbW5KOWRRZ0U2OFJpTUVaQ29mNFJnU29SQ244ZS9odnlyClJFcVA2OHVBVlU0SUVnWGlEbnc1QUhMYnJWRytSWExHbDUwcm5QSXZGMjhwbDBRa01ic293Nk5OenY4ZlhsNUYKaC94dVg1MWNBNWZnL3gvM2dQaHJQSHp2MG9iWElDQTA2VjMrRU1FK1FUQnE1UDZRYWgwZGJOWDUrVDN5MTgwdApDMzJCajJlQTBwNTNlUjFzZ0VjQ0F3RUFBYU5DTUVBd0RnWURWUjBQQVFIL0JBUURBZ0trTUE4R0ExVWRFd0VCCi93UUZNQU1CQWY4d0hRWURWUjBPQkJZRUZORlBWN0w3RlA0MzYzVkt4L0FUaWxFQnRldzFNQTBHQ1NxR1NJYjMKRFFFQkN3VUFBNElCQVFCcEY3MGt0cERBZldPc2liUDVnMy8zdlFpbFVTOUowWWdvV2JteHBqWkV6NER6YjdtOQprSVBuSFVUUG9sYW93SDBnNkQ3MTMrVGxSNW0rUzRjWGlGRzNJeGQya1BEbzFsSGtxS2JOUi93NnhsWUFyMlgwCnU0NEhjaXZtL2VMSGcrb0FYMUwvOGdYazJUUXZqWVpuNFFhU0lVRzdOZXFQTjU1NHRucW1wYVdTdXZBMG03UFgKQUVxZmk0Rm5SY0xJVGtZekFwekM4bXVxWGVmQkpHcXNDWE9yVlNCMUlIZndJcmM3M2R3K2xpTUh3dHlGYmhvSgp1TW96a0E1LzQrNU5aMGppNDl3aDJXY2VaWmwvOVNhU3BEb0V0ajFFUExTczBTSGVTd2FiNmxpVlpydUcyUHNZCm1NVGU2R1FHQWh4UDRJc0xDSHdRVjhTY1o4WklWQ0dXc1lFTQotLS0tLUVORCBDRVJUSUZJQ0FURS0tLS0tCg==\n    server: https://10.88.40.89:6443\n  name: kubernetes\ncontexts:\n- context:\n    cluster: kubernetes\n    user: kubernetes-admin\n  name: kubernetes-admin@kubernetes\ncurrent-context: kubernetes-admin@kubernetes\nkind: Config\npreferences: {}\nusers:\n- name: kubernetes-admin\n  user:\n    client-certificate-data: LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSURJVENDQWdtZ0F3SUJBZ0lJZFEvbW5NYmZnZEV3RFFZSktvWklodmNOQVFFTEJRQXdGVEVUTUJFR0ExVUUKQXhNS2EzVmlaWEp1WlhSbGN6QWVGdzB5TkRBMU1qY3dNVFEyTXpWYUZ3MHlOVEExTWpjd01UUTJNemhhTURReApGekFWQmdOVkJBb1REbk41YzNSbGJUcHRZWE4wWlhKek1Sa3dGd1lEVlFRREV4QnJkV0psY201bGRHVnpMV0ZrCmJXbHVNSUlCSWpBTkJna3Foa2lHOXcwQkFRRUZBQU9DQVE4QU1JSUJDZ0tDQVFFQXhPbFAyQ3Q5UG5GSFhMZkEKaFpneCszR3VUY29aeHhCeXZwVWxoRGsyclFMK0dvUkVkeVpRZ1V0TW9CQXFoVHdBTnNNaUNzT29ZOVBnTmxFYQpTNzE2TC9IZ2xRVjNMMENMUHJzSXdmSEdMQWxhRHFXQW42QXFZai9zY21WbDZsUGRaZ2YxS29HdkZoc3RTdFk3CjYwRkRwbEVlN2tqd2kySERNY0FmeCt1WlFISkxhK0xnYTYvWFZIMkx0QW9NV2h0Z3VvL3o2Z1paa3VMbEEzaUoKNGtYaFJabDJ6Z3pSKzIxclJSZUV6OTdPLzl5cjJmbjFVUWpQQ05zYmMyMzI4RGMxc0dsT1BkRkZvcERBNHhIYwo4STJpSmRwY3RFbDgwakpkSE9BVi9wOHEwYUhVZFFxeFlWaGJqWUFZZmdlb1RkMlNHNkJPdE1sQklsVmMxNzJDCjlaWXFod0lEQVFBQm8xWXdWREFPQmdOVkhROEJBZjhFQkFNQ0JhQXdFd1lEVlIwbEJBd3dDZ1lJS3dZQkJRVUgKQXdJd0RBWURWUjBUQVFIL0JBSXdBREFmQmdOVkhTTUVHREFXZ0JUUlQxZXkreFQrTit0MVNzZndFNHBSQWJYcwpOVEFOQmdrcWhraUc5dzBCQVFzRkFBT0NBUUVBSmlwbkg2dzZPR05OSkN2Uk0xK1RUQ1gwWUl0ZWtZQm9reWZKClNuUFdYV2tYVXhDQy9YSkl0bEMxOW1RNEZFeWlPWTVuV09SWHRZL3BHZ3BDVGFlRWNpV1ZTVGNPeHBmaURZVmUKbXZFWEFsTmRmMmNqZnhpQVBIcE0zZ0EzK0FBbTBucjlNZ3NUKysvWldyVktVQXZvV0VDcDdHQVFsempTU2haaQpXZHlaZFF6UFkyRklmT1FMV1BjUGRyQmdyR3hIbUtXUXBDV3hNV0diRTBBQU9QQXA4dXVCcURRb3pwWHRnWHo3CklnQzJQVU1CTGVKRWMzQmMxajBBc3h6OVJSWHdFcFM2bUJjWnNRQnRORElDUFhrU1VZd2VRSkZidEJ2Qm8yNlAKbEgrL05peU1WbXFZOG5oZENHZW5Bb2Z6R2JyWE95NUlYelBUdUcwK3k3NlJzdkw3WkE9PQotLS0tLUVORCBDRVJUSUZJQ0FURS0tLS0tCg==\n    client-key-data: LS0tLS1CRUdJTiBSU0EgUFJJVkFURSBLRVktLS0tLQpNSUlFb2dJQkFBS0NBUUVBeE9sUDJDdDlQbkZIWExmQWhaZ3grM0d1VGNvWnh4Qnl2cFVsaERrMnJRTCtHb1JFCmR5WlFnVXRNb0JBcWhUd0FOc01pQ3NPb1k5UGdObEVhUzcxNkwvSGdsUVYzTDBDTFByc0l3ZkhHTEFsYURxV0EKbjZBcVlqL3NjbVZsNmxQZFpnZjFLb0d2RmhzdFN0WTc2MEZEcGxFZTdrandpMkhETWNBZngrdVpRSEpMYStMZwphNi9YVkgyTHRBb01XaHRndW8vejZnWlprdUxsQTNpSjRrWGhSWmwyemd6UisyMXJSUmVFejk3Ty85eXIyZm4xClVRalBDTnNiYzIzMjhEYzFzR2xPUGRGRm9wREE0eEhjOEkyaUpkcGN0RWw4MGpKZEhPQVYvcDhxMGFIVWRRcXgKWVZoYmpZQVlmZ2VvVGQyU0c2Qk90TWxCSWxWYzE3MkM5WllxaHdJREFRQUJBb0lCQUVGdjAyUGdGKy92VHlyRgp1Y25ycVBpSmI4WlVHanJvQVduS1FMeHZFdmR0M2luTHVNRHJ6NmNYOGUvalQyNFZZWERSYWZldnk3N1pBMUtwCm94WjJjaXVLc3ZweTVZZWphR0xCWHdpc294R3hUZWMrSXI5cFI0eDg5OE9RWk16NWI4TGZkQjdiRitjOXd0QnkKMzFpdnlKNTFxTlRCRW9jUFJ3RzZlaFZDK3h2S0NiV3JqOGorM0VmRHczbGlteVlMYXg5aDd5SjcyQXBua1g3cgpYNUk5b0tQQTIybUdpVlJiRDQwTFltL2U5NGRnaEVRM3dZdDYwREtlM2ZKdSt3M0lmVWMzN3pxZDlYdWFhUmZZCjF3L0dHOExGdzRBMFJQeGJOYVkwR2lPK0V5bkJsQThTY0F3V21yek55MkRuVHc0TFFDaFZkNmEyNEJ2T0trOXQKSklYNDhYRUNnWUVBeW0rNmxlV2ZjdWpFN05lb09sU0o4SkJ0eSthODQ2V0pkWlBuVGhXL3hWdnJ4Z3llRjJ0cgp5UENPZHZBS1NFVEN4OGdBOWVCbFYwQkp5Z29iK0RBRm5LWmxRcnVqa2MyTVVJMUpuVXoyVmVEOW5VZnpJM3dLCjh2RlZFTmxld1JZcllrSklXbmd4dmVsSkRyWWZIQTN2V2M1c29ZbnI0WGpLWEcvaG5TUmorQmtDZ1lFQStRTlcKRHJ2cVVQNit4dTRWVnZmMktQVmcra3QrcDdSTVFuNHErZFpvTWdTNE9ISEM4M2VRZHFOQmNNTHUyZjdMWUJzdwpGL3NCOURMNjE3eldJZ0ZPbFdNTm5wdmZISENsdktGUktsZ3lMZys4NE5SbTdxaFp1VUxJMVNwcnlidVlVeGFmCit0Z0JWUzZxRUtZblB4enBGTFBHaVY5K3RKcUhnbGdYU0NGR0M1OENnWUFza1FSbTA3S1N2YVA4OEUxYzZoZWsKR21mUnArNlZ2Mi9QcnN3UmFNQmJWcTFGWUttOVJpem9nKzBaU2ljNVdjbjM2WlFzalVaQ095a0RqYzR5aVgveQp5TDhxUGR6NDBHTy9xQnM0d2lUSkswc3FRUmpLQ3hqdCt5dGJGai9DQ250cW1tVUxaY0t0cUpvbzN2ZGlXYjhECjZtb0g5akV6NWovR3NlcmRpWkZvb1FLQmdGNGtaYit5U2JjUGFINzYrZlJrVUp0UmtiVnZIMVkxOHZMY2tsRGMKUDQ1UmZIaFNnaHJiNURFR0xiSjg4Q0pZa3ZrbHJqd1N1TDFqSExTOWZ0TmlYUnhiVTRYWWRqUmhYamQvWVprNQp1TmJ6T2p4aDhQeGZoM1JrREZ1UzZ5ZGNsZzBycTVZQlhoZWZTY3A3bHJEa3REWVFkWW53SVk0VU1oRVVUdEVGClMxOFBBb0dBQmRJTndDMXhRZVlvbUFaQjVoVXk0bUtwZHhCSjBsekthWGJ0UnNRQ3ZqdHN5alhJemZQUVlEMmQKZnlGdFcyTTM3K0V5elFRTWRNYVhpYXM5V2V5dTdRS2JjckVIZXE4RmlEV3FtUFRoQjk4VE5XZ3ozQ3dRZ092TQpNQXdsZkFWVWdxWUZxWXorMFYydkFyMVdpRXdFS2dRRld4VlpqUUJDUnZGZ2lKaWlsZEU9Ci0tLS0tRU5EIFJTQSBQUklWQVRFIEtFWS0tLS0tCg==";

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        log.info("[ After Connection Established ]： 开始处理webSocket");
        // 获取客户端传递的参数
        URI uri = session.getUri();
        String query = uri.getQuery();
        String namespace = "default";
        String podName = "";
        String containerName = "";
        String kubeConfig = "";
        int sinceSeconds = 0;
        int tailLines = 500;

        if (query != null) {
            Map<String, String> params = parseQuery(query);

            namespace = params.getOrDefault("namespace", namespace);
            podName = params.getOrDefault("podName", podName);
            containerName = params.getOrDefault("containerName", containerName);
            sinceSeconds = Integer.parseInt(params.getOrDefault("sinceSeconds", String.valueOf(sinceSeconds)));
            tailLines = Integer.parseInt(params.getOrDefault("tailLines", String.valueOf(tailLines)));
            kubeConfig = URLDecoder.decode(params.getOrDefault("kubeConfig", kubeConfig), StandardCharsets.UTF_8.toString());
        }

        // 填充请求
        PodQueryDTO queryDTO = new PodQueryDTO();
        queryDTO.setNamespaceName(namespace);
        queryDTO.setContainer(containerName);
        queryDTO.setPodName(podName);
        // 仅当sinceSeconds不为0时设置
        if(sinceSeconds != 0) {
            queryDTO.setSinceSeconds(sinceSeconds);
        }
        queryDTO.setTailLines(tailLines);
        queryDTO.setKubeConfig(kubeConfig);

        // 启动日志流任务
        podService.streamPodLog(queryDTO, session);

        // 启动心跳机制
        heartbeatExecutorService = Executors.newScheduledThreadPool(1);
        heartbeatExecutorService.scheduleAtFixedRate(() -> {
            try {
                if (session.isOpen()) {
                    session.sendMessage(new TextMessage("ping"));
                    log.info("Sent ping to client");
                    missedHeartbeats = 0; // Reset missed heartbeats on successful ping
                }
                missedHeartbeats++;

                if (missedHeartbeats > MAX_MISSED_HEARTBEATS) {
                    log.warn("Missed too many heartbeats, closing session");
                    session.close(CloseStatus.GOING_AWAY);
                    shutdownHeartbeatExecutor();
                }
            } catch (IOException e) {
                log.error("Error sending ping to client: " + e.getMessage(), e);
                shutdownHeartbeatExecutor();
            }
        }, 0, 30, TimeUnit.SECONDS);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.info("Connection closed: " + status);
        log.info("[ After Connection Established ]： 停止处理webSocket");
        shutdownHeartbeatExecutor();
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        log.error("Transport error: " + exception.getMessage(), exception);
        shutdownHeartbeatExecutor();
    }

    private void shutdownHeartbeatExecutor() {
        if (heartbeatExecutorService != null && !heartbeatExecutorService.isShutdown()) {
            heartbeatExecutorService.shutdownNow();
        }
    }

    // Function to parse query string into a map
    private Map<String, String> parseQuery(String query) {
        Map<String, String> params = new HashMap<>();
        String[] keyValuePairs = query.split("&");
        for (String pair : keyValuePairs) {
            int idx = pair.indexOf("=");
            String key = idx > 0 ? pair.substring(0, idx) : pair;
            String value = idx > 0 && pair.length() > idx + 1 ? pair.substring(idx + 1) : "";
            params.put(key, value);
        }
        return params;
    }

}